package org.fujene.communicate.repli;

import java.net.UnknownHostException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.TreeSet;

import org.fujene.structure.Constants;
import org.fujene.structure.ContentInfo;
import org.fujene.structure.ExtendQueue;
import org.fujene.structure.LookupPointer;
import org.fujene.structure.Node;
import org.fujene.structure.ReplicateInfo;
import org.msgpack.MessagePackObject;
import org.msgpack.object.ArrayType;
import org.msgpack.object.BooleanType;
import org.msgpack.object.IntegerType;
import org.msgpack.object.RawType;
import org.msgpack.rpc.Client;

/**
 * Messagepack-RPC ノード間通信のためのメソッドを定義します。
 * @author iamyusk
 */
public class ReplicateCommunicator{
	private static final int PEERPORT = Constants.PEER_PORT;
	private static final int BUFFERSIZE = 100;
	private static final int OUTER = 0;
	private static final int INNER = 1;

	public ReplicateInfo getInfo(String host){
		Client client = null;
		ReplicateInfo retInfo = null;

		try{
			client = new Client(host, PEERPORT);
			client.setRequestTimeout(3);

			// 情報取得
			MessagePackObject infoobj = client.callApply("getInfo", new Object[]{});
			if(infoobj.isArrayType()){
				MessagePackObject[] infos = infoobj.asArray();
				String[] contStr = new String[infos[4].asInt()];
				String[] appStr = new String[infos[5].asInt()];
				byte[] graph = infos[6].asByteArray();
				for(int i = 0; i < contStr.length; i++)
					contStr[i] = infos[7 + i].asString();
				for(int i = 0; i < appStr.length; i++)
					appStr[i] = infos[7 + i + contStr.length].asString();

				ContentInfo contentInfo = new ContentInfo(contStr, appStr);
				contentInfo.setConnection(graph);
				retInfo = new ReplicateInfo(infos[0].asInt(), infos[1].asInt(), infos[2].asInt(),
						infos[3].asInt(), contentInfo);
			}else{
				System.out.println("getInfo did not called.");
			}
		}catch(Exception e){
			e.printStackTrace();
		}finally{
			if(client != null) client.close();
		}

		return retInfo;
	}

	public TreeSet<Node> getNodeList(String host, int type) throws Exception{
		TreeSet<Node> nodeList = new TreeSet<Node>();
		Client client = null;
		Node newNode;
		try{
			client = new Client(host, PEERPORT);
			client.setRequestTimeout(5);

			MessagePackObject nodes = client.callApply("nodeList", new Object[]{ type });
			if(nodes.isArrayType()){
				MessagePackObject[] objs = nodes.asArray();
				for(int i = 0; i < objs.length; i += 2){
					newNode = new Node(objs[i].asString(), objs[i + 1].asLong());
					newNode.indexable();
					nodeList.add(newNode);
				}
			}
		}finally{
			if(client != null) client.close();
		}

		return nodeList;
	}

	public void register(String host, String myAddress, long myOuterHash, long myInnerHash){
		Client client = null;

		try{
			client = new Client(host, PEERPORT);
			client.setRequestTimeout(5);

			client.callApply(
					"register",
					new Object[]{ RawType.create(myAddress), IntegerType.create(myOuterHash),
							IntegerType.create(myInnerHash), });
		}catch(UnknownHostException e){
			e.printStackTrace();
		}finally{
			if(client != null) client.close();
		}
	}

	public MessagePackObject[] makeRemoteContentObject(ExtendQueue contents){
		MessagePackObject[] contentObjs = new MessagePackObject[contents.contentTypes
				+ contents.appendixTypes];
		for(int i = 0; i < contents.contentTypes; i++)
			contentObjs[i] = RawType.create(contents.contents[i]);

		for(int i = 0; i < contents.appendixTypes; i++)
			contentObjs[i + contents.contentTypes] = RawType.create(contents.appendices[i]);

		return contentObjs;
	}

	public void indexContentRemote(ArrayType contents, long ID, Node node) throws Exception{
		Client client = null;

		try{
			client = new Client(node.getAddr(), PEERPORT);

			client.callApply("indexContent", new Object[]{ contents, IntegerType.create(ID) });
		}finally{
			if(client != null) client.close();
		}
	}

	public void indexTermRemote(String term, int termfreq, long ID, int type, Node node) throws Exception{
		Client client = null;

		try{
			client = new Client(node.getAddr(), PEERPORT);

			client.callApply("indexTerm", new Object[]{ RawType.create(term), IntegerType.create(termfreq),
					IntegerType.create(ID), IntegerType.create(type) });
		}finally{
			if(client != null) client.close();
		}
	}

	public byte[] searchRemote(String newQuery, int type, Node node) throws Exception{
		Client client = null;
		byte[] retb = null;

		try{
			client = new Client(node.getAddr(), PEERPORT);

			MessagePackObject ret = client.callApply("searchTerm", new Object[]{ RawType.create(newQuery),
					IntegerType.create(type) });

			if(ret.isRawType()) retb = ret.asByteArray();
		}finally{
			if(client != null) client.close();
		}

		return retb;
	}

	/**
	 * 単語リストを取得します(第一段階)。
	 * 取得するのは単語ハッシュiが hashFrom <= i <= hashTo となるものに限ります。
	 * @param node ... ノード
	 * @param hashFrom ... ハッシュ開始位置
	 * @param hashTo ... ハッシュ終了位置
	 * @param myAddress ... 自ノードアドレス
	 * @return <単語, <転置インデックスへのポインタ>>
	 */
	public HashMap<String, ArrayList<LookupPointer>> getTermList(Node node, long hashFrom, long hashTo,
			String myAddress){
		Client client = null;
		int indexNum = 0;
		HashMap<String, ArrayList<LookupPointer>> retMap = new HashMap<String, ArrayList<LookupPointer>>();
		ArrayList<LookupPointer> ptrList = null;
		MessagePackObject[] objs = null;

		try{
			client = new Client(node.getAddr(), PEERPORT);
			client.setRequestTimeout(50);

			MessagePackObject obj = client.callApply("getDataSize", new Object[]{ IntegerType.create(OUTER),
					IntegerType.create(hashFrom), IntegerType.create(hashTo), RawType.create(myAddress),
					BooleanType.create(true) });

			if(obj.isIntegerType()){
				indexNum = obj.asInt();
				System.out.println("Term: Read " + indexNum + " records.");
			}

			for(int i = 0; i < indexNum; i++){
				obj = client.callApply("getData",
						new Object[]{ IntegerType.create(i), RawType.create(myAddress) });

				if(obj.isArrayType()){
					objs = obj.asArray();
					ptrList = new ArrayList<LookupPointer>();

					for(int j = 0; 3 * j + 3 < objs.length; j++)
						ptrList.add(new LookupPointer(objs[3 * j + 2].asInt(), objs[3 * j + 3].asInt(),
								objs[3 * j + 1].asLong()));

					retMap.put(objs[0].asString(), ptrList);
				}
			}

			client.callApply("endDataSend", new Object[]{ RawType.create(myAddress) });
		}catch(Exception e){
			e.printStackTrace();
		}finally{
			if(client != null) client.close();
		}

		return retMap;
	}

	/**
	 * 単語リストを取得します(第二段階)。
	 * @param host
	 * @param ptr
	 * @return
	 */
	public LinkedHashMap<Long, Integer> getInvertIndexList(Node node, LookupPointer ptr){
		LinkedHashMap<Long, Integer> invertIndices = new LinkedHashMap<Long, Integer>();
		Client client = null;

		try{
			client = new Client(node.getAddr(), PEERPORT);
			while(!ptr.equals(Constants.EMPTY_PTR)){
				// 転置インデックス取得
				MessagePackObject hashObj = client.callApply("getList",
						new Object[]{ IntegerType.create(ptr.block), IntegerType.create(ptr.point),
								IntegerType.create(ptr.firstID), IntegerType.create(BUFFERSIZE) });
				if(hashObj.isArrayType()){
					MessagePackObject[] hashObjs = hashObj.asArray();
					if(hashObjs.length >= 3){
						// FIXME: Index中だとエラーっぽい
						ptr.block = hashObjs[0].asInt();
						ptr.point = hashObjs[1].asInt();
						ptr.firstID = hashObjs[2].asLong();
						for(int k = 3; k + 1 < hashObjs.length; k += 2){
							invertIndices.put(Long.valueOf(hashObjs[k].asLong()),
									Integer.valueOf(hashObjs[k + 1].asInt()));
						}
					}
				}
			}
		}catch(Exception e){
			e.printStackTrace();
		}finally{
			if(client != null) client.close();
		}

		return invertIndices;
	}

	/**
	 * RPC経由で本文リストを取得します。
	 * @param host
	 * @param hashFrom
	 * @param hashTo
	 * @param myAddress
	 * @param contentInfo
	 * @return
	 */
	public HashSet<ExtendQueue> getContentListFromNode(Node node, long hashFrom, long hashTo,
			String myAddress, ContentInfo contentInfo){
		Client client = null;
		int indexNum = 0;
		HashSet<ExtendQueue> retList = new HashSet<ExtendQueue>();

		try{
			client = new Client(node.getAddr(), PEERPORT);
			client.setRequestTimeout(50);

			MessagePackObject obj = client.callApply("getDataSize", new Object[]{ IntegerType.create(INNER),
					IntegerType.create(hashFrom), IntegerType.create(hashTo), RawType.create(myAddress),
					BooleanType.create(false) });
			MessagePackObject[] objs;

			if(obj.isIntegerType()){
				indexNum = obj.asInt();
				System.out.println("Content: Read " + indexNum + " records.");
			}

			for(int i = 0; i < indexNum; i++){
				obj = client.callApply("getData",
						new Object[]{ IntegerType.create(i), RawType.create(myAddress) });

				if(obj.isArrayType()){
					objs = obj.asArray();
					long ID = objs[0].asLong();
					String[] contents = new String[contentInfo.getContentNum()];
					String[] supplements = new String[contentInfo.getAppendixNum()];
					for(int j = 0; j < contents.length; j++)
						contents[j] = objs[j + 1].asString();
					for(int j = 0; j < supplements.length; j++)
						supplements[j] = objs[j + 1 + contents.length].asString();

					retList.add(new ExtendQueue(contents, supplements, ID));
				}
			}

			client.callApply("endDataSend", new Object[]{ RawType.create(myAddress) });
		}catch(Exception e){
			e.printStackTrace();
		}finally{
			if(client != null) client.close();
		}

		return retList;
	}

	public ExtendQueue
			getContentSet(long ID, int priorIndex, int contentLength, int appendixLength, Node node){
		Client client = null;
		ExtendQueue retVal = null;
		int i;

		try{
			client = new Client(node.getAddr(), PEERPORT);
			client.setRequestTimeout(3);

			MessagePackObject obj = client.callApply("getContent", new Object[]{ IntegerType.create(ID),
					IntegerType.create(priorIndex) });
			if(obj.isArrayType()){
				MessagePackObject[] array = obj.asArray();
				String[] peerContent = new String[contentLength];
				String[] peerSupplement = new String[appendixLength];

				for(i = 0; i < peerContent.length; i++)
					peerContent[i] = array[i].asString();
				for(i = 0; i < peerSupplement.length; i++)
					peerSupplement[i] = array[i + peerContent.length].asString();

				retVal = new ExtendQueue(peerContent, peerSupplement, ID);
			}
		}catch(Exception e){
			e.printStackTrace();
		}finally{
			if(client != null) client.close();
		}

		return retVal;
	}

	public boolean isLive(Node node) throws Exception{
		Client client = null;
		MessagePackObject result;
		boolean retb = false;

		try{
			client = new Client(node.getAddr(), PEERPORT);
			client.setRequestTimeout(3);

			result = client.callApply("isLive", new Object[]{});
			if(result.isRawType()) retb = true;
		}finally{
			if(client != null) client.close();
		}

		return retb;
	}

	public void iWillDie(Node node){
		Client client = null;
		try{
			client = new Client(node.getAddr(), PEERPORT);

			client.callApply("delete", new Object[]{ 0 });
			client.callApply("delete", new Object[]{ 1 });
		}catch(Exception e){}finally{
			if(client != null) client.close();
		}
	}

	public void deleteContent(Node node, long ID){
		Client client = null;
		try{
			client = new Client(node.getAddr(), PEERPORT);

			client.callApply("deleteContent", new Object[]{ IntegerType.create(ID) });
		}catch(Exception e){}finally{
			if(client != null) client.close();
		}
	}

	public void indexable(Node node, Node myNode){
		Client client = null;
		try{
			client = new Client(node.getAddr(), PEERPORT);

			client.callApply("callIndexable",
					new Object[]{ RawType.create(myNode.getAddr()), BooleanType.create(true) });
		}catch(Exception e){}finally{
			if(client != null) client.close();
		}
	}
}
